package ru.dadle8;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.List;

public class SampleConsumer extends AbstractConsumer {

    @Override
    public void readMessages(String topicName) {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(getKafkaProperties());

        //Kafka Consumer subscribes list of topics here.
        consumer.subscribe(List.of(topicName));
        consumer.seekToBeginning(consumer.assignment());
        //print the topic name
        System.out.println("Subscribed to topic " + topicName);

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, String> record : records)

                // print the offset,key and value for the consumer records.
                System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());

            // to read from beginning
            // Now there is heartbeat and consumer is "alive"
//             consumer.seekToBeginning(consumer.assignment());
        }
    }

    public static void main(String[] args) {
        String topicName = "send-order-event";
        SampleConsumer sampleConsumer = new SampleConsumer();
        sampleConsumer.readMessages(topicName);
    }
}
